package scau.cxq.homework;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.HashSet;
import java.util.Set;


public class ExecuteMapper extends Mapper<LongWritable, Text, Text, NullWritable> {

    private static final Set<String> pages = new HashSet<String>();

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        pages.add("/about");
        pages.add("/black-ip-list/");
        pages.add("/cassandra-clustor/");
        pages.add("/finance-rhive-repurchase/");
        pages.add("/hadoop-family-roadmap/");
        pages.add("/hadoop-hive-intro/");
        pages.add("/hadoop-zookeeper-intro/");
        pages.add("/hadoop-mahout-roadmap/");

    }

    @Override
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {

        Text k = new Text();
        NullWritable v = NullWritable.get();

        // 把一行内容转成字符串
        String lineStr = value.toString();
        ExecuteItem executeItem = ExecuteItem.parse(lineStr);

        if (executeItem == null) {
            return;
        }

        // 过滤
        filtStaticResource(executeItem);

//        System.out.println(executeItem.toString());
        k.set(executeItem.toString());

        context.write(k, v);
    }

    public static void filtStaticResource(ExecuteItem item) {
        if (!pages.contains(item.getRequest())) {
            item.setValid(false);
        }
    }

    private boolean isValid(String s) {

        for (String path : pages) {
            if (s.contains(path)) {
                return true;
            }
        }
        return false;
    }
}
